#include <utime.h>
#include <dirent.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <iostream>
#include <jasper/jasper.h>
#include "warpUtil.h"
#include "warpsol.h"


/** Some helper routines for warpUtil object */
typedef enum  {
  ENOTREAD, 
  ENOTWRITE, 
  EEXIST, 
  EDIREXIST,
  EOPENDIR, 
  EFILETYPECOMPRESS, 
  EFILETYPEUNCOMPRESS, 
  ESPECIAL,
  ESTAT,
  ETIME,
  ESPERM,
  EMKDIR,
  EFORK
} MSG;

static const char messages[][128] = {
  "Can not read input file %s\n",
  "Can not write output file %s\n",
  "Cowardly refusing to overwrite existing image %s\nUse -o option to force overwrite or -C to ignore and leave untouched\n",
  "Target directory %s exists. Please choose a non-existent target directory when using -D\n",
  "Can not open directory %s for reading\n",
  "Can only convert .tif files while attempting to compress %s. File ignored and not cmpressed.\n",
  "Can only convert .jpc files while attempting to uncompress %s. File ignored and not uncompressed.\n",
  "Not a regular file or directory: %s, ignored\n",
  "Can not stat file or directory %s\n",
  "Can not adjust access times for %s\n"
  "Can not adjust permissions for %s\n"
  "Can not create directory %s\n"
};

static void message(MSG i, string fn) {
  int needMem = strlen(messages[i]) + fn.length() + 1;
  char *buffer = new char[needMem];
  sprintf(buffer, messages[i], fn.c_str());
  std::cerr << buffer;
  delete [] buffer;
}



int preserveTimeAndPerm(const string &templ, const string &target) {
    struct stat sstat;
    int retval = 0;
    if(stat(templ.c_str(), &sstat)) {
      message(ESTAT, target.c_str());
      retval =  -ESTAT;
    } 
    if(chmod(target.c_str(), sstat.st_mode)) {
      message(ESPERM, target.c_str());
      retval =  -ESPERM;
    } 
    struct utimbuf T;
    T.actime = sstat.st_atime;
    T.modtime  = sstat.st_mtime;
    if(utime(target.c_str(), &T)) {
      message(ESTAT, target.c_str());
      retval= -ETIME;
    } 
    return retval;
}

static bool fileExists(string fn) {
  FILE *fd;
  if( (fd = fopen(fn.c_str(), "r")) ) {
    fclose(fd);
    return true;
  }
  return false;
}

/**********************************************************/
/* Warputil object implementation                         */
/**********************************************************/
warputil::warputil() : _coward(true), _rate("1.0")
#if defined HAVE_THREADS || defined USE_FORK
		     ,_nproc(1)
#endif
   {jas_init();}; 

warputil::~warputil() {jas_cleanup();  }

int warputil::compressFile(string infile, string outfile, warpSol &warp) {
  if(fileExists(outfile)) {
    if(continuation()) return 0;
    if(coward()) {
      message(EEXIST, outfile);
      return -EEXIST;
    }
  }
  string r = "rate=" + rate();
  if(tilesize() != "") r +=  " tilewidth=" + tilesize() + " tileheight=" + tilesize();
  if(warp.readSolexa(infile)) {
    message(ENOTREAD, infile);
    return -ENOTREAD;
  }
  if(warp.writeJasper(outfile, r.c_str(), "jpc")) {
    message(ENOTWRITE, outfile);
    return -ENOTWRITE;
  }

  if(preserveAP()) preserveTimeAndPerm(infile, outfile);
  return 0;
}


int warputil::unCompressFile(string infile, string outfile, warpSol &warp) {
  if(fileExists(outfile)) {
    if(continuation()) return 0;
    if(coward()) {
      message(EEXIST, outfile);
      return -EEXIST;
    }
  }

  if(warp.readJasper(infile.c_str())) {
    message(ENOTREAD, infile);
    return -ENOTREAD;
  }

  if(warp.writeSolexa(outfile.c_str())) {
    message(ENOTWRITE, outfile);
    return -ENOTWRITE;
  }
  if(preserveAP()) preserveTimeAndPerm(infile, outfile);
  return 0;
}

int warputil::_crawl(string sourceDir, string targetDir, bool compress) {
  DIR *D = opendir(sourceDir.c_str());
  int errcode = 0;
  if(D) {
    if(DIR *T = opendir(targetDir.c_str())) {
      closedir(T);
      if(!continuation()) {
	message(EEXIST, targetDir);
	errcode = -EEXIST;
	goto leave;
      }
    } else {
      if(mkdir(targetDir.c_str(), S_IRWXU | S_IXGRP | S_IRGRP)) {
	message(EMKDIR, targetDir);
	errcode =  -EMKDIR;
	goto leave;
      }
    }


    while(dirent *cur = readdir(D)) {
      struct stat info;
      string curnam = sourceDir + "/" + cur->d_name;
      if(stat(curnam.c_str(), &info)) {
	message(ESTAT, curnam);
	errcode = -ESTAT;
	goto leave;
      }

      if(S_ISDIR(info.st_mode)) {
	if(! (strcmp(cur->d_name, ".") && strcmp(cur->d_name, ".."))) continue;
	string subdir    = sourceDir  + "/" + cur->d_name + "/";
	string subtarget = targetDir  + "/" + cur->d_name;
	errcode = _crawl(subdir, subtarget, compress);
	if(errcode < 0) goto leave;
	continue;
      }
      if(S_ISREG(info.st_mode)) {
	string ifnam    = sourceDir + "/" + cur->d_name;
	string ofnam    = targetDir + "/" + cur->d_name;
	string iext     = compress ? ".tif" : ".jpc";
	string oext     = compress ? ".jpc" : ".tif";
	if(ifnam.substr(ifnam.length()-4) != iext) {
	  message(EFILETYPECOMPRESS, ifnam);
	  continue;
	}
	ofnam =  ofnam.substr(0, ofnam.length()-4) + oext;
	_schedule(ifnam, ofnam, compress);
	continue;
      }
      message(ESPECIAL, cur->d_name);
    }
    //    if(preserveAP()) preserveTimeAndPerm(sourceDir, targetDir);

  leave:
    closedir(D);
    return errcode;

  } else {
    message(EOPENDIR, sourceDir);
    return -EOPENDIR;
  }
}


#ifndef HAVE_THREADS
int warputil::processDirs(string sourceDir, string targetDir, bool compress) {
  _crawl(sourceDir, targetDir, compress); // Identify Jobs
  const int L = _pipeline.size();
#ifndef USE_FORK
    return _executePipeline(0, L);
#else 
  int retval = 0;
  vector<pid_t> pids(nProcessor());
  int chunksize = L / nProcessor();
  // Fork worker processes
  for(int i=0; i < nProcessor(); ++i) {
    pids[i] = fork();
    if(pids[i] < 0) return -EFORK;
    if(!pids[i]) {
      int start = i * chunksize;
      int end   = (i == nProcessor() - 1) ? L : start + chunksize;
      int r = _executePipeline(start, end);
      exit(r); // Terminate worker process
    }
  }
  
  // wait for worker processes to terminate
  int running = nProcessor();
  while(running) {
    siginfo_t infop;
    waitid(P_ALL, 0, &infop, WEXITED);
    for(int i=0;  i < nProcessor(); ++i) { if(pids[i] == infop.si_pid) { pids[i] = -1; break; }}
    if(infop.si_status != 0) {
      retval = infop.si_status;
      // Error occured, kill other processes
      for(int i=0; i < nProcessor(); i++) {
	if(pids[i] >= 0) {
	  kill(pids[i], SIGKILL); // Its a little brutal, is there any more graceful methods ?
	}
      }
      return retval;
    }
    --running;
  }
  return retval;
#endif // USE_FORK
}

void warputil::_schedule(const string &in, const string &out, bool mode) {
  _pipeline.push_front(_task(in, out, mode));
}


int warputil::_executePipeline(int start, int end) {
  for(int i=start; i < end; ++i) {
    warpSol warp;
    _task &job = _pipeline[i];
    int retval = job.compress() ? compressFile(job.in(), job.out(), warp) : unCompressFile(job.in(), job.out(), warp);
    if(retval && !robust()) return retval;
  }
  return 0;
}

#else
int warputil::processDirs(string sourceDir, string targetDir, bool compress) {
  _asyncStart();
  int r = _crawl(sourceDir, targetDir, compress);
  _asyncEnd();
  return r ? r : _errorcode;
}

void warputil::_schedule(const string &in, const string &out, bool mode) {
  pthread_mutex_lock(&_mutex);
  _pipeline.push_front(_task(in, out, mode));
  pthread_cond_signal(&_master);
  pthread_mutex_unlock(&_mutex);
}

void *slaveEntryPoint(void *arg) {
  warputil &object = *static_cast<warputil *>(arg);
  object._aSlavesLife();
  return NULL;
}


void warputil::_asyncStart() {
  _errorcode = 0;
  _finalize = _cancel = false;
  pthread_mutex_init(&_mutex, 0);
  pthread_cond_init(&_slave, 0);
  pthread_cond_init(&_master, 0);
  _slaves.resize(nProcessor());
  for(int i = 0; i < nProcessor(); ++i) {
    pthread_create(&_slaves[i], NULL, slaveEntryPoint, (void *) this);
  }
  return;
}

void warputil::_asyncEnd() {
    pthread_mutex_lock(&_mutex);
    _finalize = true;

    while(1) {
      if(_errorcode && !_robust) {
	_cancel = true;
	pthread_cond_broadcast(&_master);
	break;
      }
      if(!_pipeline.size()) break; // We are done
      pthread_cond_wait(&_slave, &_mutex); 
    }
    pthread_cond_broadcast(&_master); // wakeup sleeping threads
    pthread_mutex_unlock(&_mutex);
    for(int i=0; i < nProcessor(); ++i) {
      pthread_join(_slaves[i], 0);
    }
    pthread_cond_destroy(&_slave);
    pthread_cond_destroy(&_master);
    pthread_mutex_destroy(&_mutex);
    _slaves.resize(0);
    return;
}

void warputil::_aSlavesLife() {
  pthread_mutex_lock(&_mutex);
  while(1) {
    if(_cancel) { // Shutdown requested
      pthread_mutex_unlock(&_mutex);
      pthread_exit(0);
    }
    
    const int L = _pipeline.size();
    if(L) {
      _task job = _task(_pipeline[L-1]);
      _pipeline.pop_back();
       // Allow other threads to run simultaneously
      pthread_mutex_unlock(&_mutex);
      warpSol warp;
      int retval = job.compress() ? compressFile(job.in(), job.out(), warp) : unCompressFile(job.in(), job.out(), warp);
      pthread_mutex_lock(&_mutex);

      if(retval) { // signal error event
	_errorcode = retval;
	pthread_cond_signal(&_slave);
      } 
      continue; // back to production
    }

    if(_finalize) { // Exit requested if no jobs are left
      pthread_cond_signal(&_slave);
      pthread_mutex_unlock(&_mutex);
      pthread_exit(0);
    }

    //wait for new jobs  
    pthread_cond_wait(&_master, &_mutex);
  } 
}
#endif // HAVE_THREADS

